package light.network;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousDatagramChannel;
import java.nio.channels.CompletionHandler;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

public class UdpEndpoint implements Endpoint {

	int bufferSize = 1 * 1024;
	AsynchronousDatagramChannel server;
	IEndpointHandler handler;

	int listenerLevel = 1;
	Deque<UdpEndpointSession> completes = new ArrayDeque<UdpEndpointSession>(1024);
	Map<String, TimeUnit> timeoutExpections = new HashMap<String, TimeUnit>();
	private TimeUnit getTimeoutUnit(SocketAddress address)
	{
		TimeUnit unit = timeoutExpections.get(genEndpointId(address));
		return unit;
	}
	private TimeUnit getTimeoutUnitSure(SocketAddress address)
	{
		TimeUnit unit = timeoutExpections.get(genEndpointId(address));
		if(unit==null)
		{
			unit = new TimeUnit();
			timeoutExpections.put(genEndpointId(address), unit);//System.out.println("put:"+genEndpointId(address));
		}
		return unit;
	}
	Map<Integer, ActUnit> acts = new HashMap<Integer, ActUnit>();

	boolean processQueue = true;

	public UdpEndpoint() throws IOException {
		server = AsynchronousDatagramChannel.open();// null, group
	}

	public void sendTo(Packet pack, SocketAddress target) {
		UdpPacket p = ((UdpPacket) pack);
		if (p.isNeedAct()) {
			synchronized (acts) {
				server.send(pack.buffer(), target);
				ActUnit u = new ActUnit();
				u.pack = pack;
				u.target = target;
				final TimeUnit tu = getTimeoutUnitSure(target);
				u.ackTime = tu.avgTime;
				if(tu.lastPackId==-1)//last pack is finished
				{
					tu.lastPackId = p.getId();
					tu.lastSendTime = System.currentTimeMillis();
				}
				acts.put(((UdpPacket) pack).getId(), u);
			}
		} else
			server.send(pack.buffer(), target);
	}

	private final void sendToUnack(Packet pack, SocketAddress target) {
		server.send(pack.buffer(), target);
	}

	public void sendTo(Packet pack, SocketAddress target, Object attachment, CompletionHandler<Integer, Object> h) {
		if (((UdpPacket) pack).isNeedAct()) {
			synchronized (acts) {
				server.send(pack.buffer(), target, attachment, h);
				ActUnit u = new ActUnit();
				u.pack = pack;
				u.target = target;
				u.handler = h;
				TimeUnit tu = getTimeoutUnitSure(target);
				u.ackTime = tu.avgTime;
				tu.lastSendTime = System.currentTimeMillis();
				acts.put(((UdpPacket) pack).getId(), u);
			}
		} else
			server.send(pack.buffer(), target, attachment, h);
	}

	private final void sendToUnack(Packet pack, SocketAddress target, Object attachment, CompletionHandler<Integer, ? super Object> h) {
		server.send(pack.buffer(), target, attachment, h);
	}

	private final void resend(Packet pack, SocketAddress target, Object attachment, CompletionHandler<Integer, ? super Object> h) {
		pack.flip();
		if (h == null)
			sendToUnack(pack, target);
		else
			sendToUnack(pack, target, attachment, h);
	}

	private void start() {
		// 监听线程 keep-alive
		if (listenerLevel == 1) {
			ComplHandler h = new ComplHandler();
			server.receive(h.buf, server, h);
		} else {
			for (int i = 0; i < listenerLevel; i++) {
				final ComplHandler h = new ComplHandler();
				new Thread() {
					@Override
					public void run() {
						server.receive(h.buf, server, h);
					};
				}.start();
			}
		}

		// 处理线程 keep-alive
		new Thread() {
			@Override
			public void run() {
				while (processQueue) {
					UdpEndpointSession s = completes.pollFirst();
					if (s == null) {
						try {
							Thread.sleep(100);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					} else {
						if (handler != null)
							handler.onReceive(s, s.getDataBuffer());
					}
					Thread.yield();
				}
			};
		}.start();
		Timer t = new Timer();
		t.schedule(new TimerTask() {
			@Override
			public void run() {
				synchronized (acts) {
					Iterator<Integer> iter = acts.keySet().iterator();
					while (iter.hasNext()) {
						int x = iter.next();
						ActUnit u = acts.get(x);
						if (u.ackTime <= 0)// 超时重发
						{
							resend(u.pack, u.target, u.attachment, u.handler);
							// 重置超时等待时间
							u.ackTime = getTimeoutUnit(u.target).avgTime;
							u.retry--;
							if (u.retry <= 0)// 超过重复次数，放弃
								iter.remove();
						} else
							u.ackTime--;
					}
				}
			}
		}, 1000, 200);
	}

	@Override
	public void start(SocketAddress address) throws IOException {
		server.bind(address);
		start();
	}

	@Override
	public void bind(SocketAddress address) throws IOException {
		server.bind(address);
		start();
	}

	@Override
	public void stop() {
		processQueue = false;
	}

	private final UdpEndpoint self() {
		return this;
	}

	//static AtomicInteger aaa = new AtomicInteger();
	class ComplHandler implements CompletionHandler<SocketAddress, Object> {
		public ByteBuffer buf = ByteBuffer.allocate(bufferSize);
		@Override
		public void completed(SocketAddress result, Object a) {
			// 马上进行下一次接收--------------------
			ComplHandler h = new ComplHandler();
			server.receive(h.buf, a, h);
			// -------------------------------------
			buf.flip();

			// 判断数据类型
			if (buf.limit() <= 0)
				return;
			UdpPacket dst = UdpPacket.wrap(buf);
			if (dst.isActPacket()) {//System.out.println("recv ack: "+aaa.incrementAndGet());
				int packId = dst.getId();
				synchronized (acts) {
					acts.remove(packId);
				}
				TimeUnit unit = getTimeoutUnit(result);
				if(unit!=null)
				{
					int elapsedTime  = (int) (System.currentTimeMillis()-unit.lastSendTime);
					if(unit.lastPackId==packId)
					{
						unit.avgTime = (unit.avgTime+elapsedTime)/2;
						unit.lastPackId = -1;
					}
					else if(elapsedTime>TimeUnit.MaxTime)
					{
						unit.lastPackId = -1;
					}
				}
				return;
			}

			// 开始处理上一次完成接收的数据
			if (!dst.validate())// 数据包长度不正确，忽略改包
			{
				failed(new Exception("package format error."), a);
				return;
			}
			UdpEndpointSession session = new UdpEndpointSession(self(), result);
			session.setDataBuffer(dst);
			session.setAttachment(a);
			completes.addLast(session);
			if(dst.isNeedAct())
			{
				UdpPacket ac = UdpPacket.newActPack(dst.getId());
				sendToUnack(ac, result);
			}
		}

		@Override
		public void failed(Throwable exc, Object server) {
			if (handler != null)
				handler.onFail(null, exc);
		}
	}

	@Override
	public int getBufferSize() {
		return bufferSize;
	}

	@Override
	public void setBufferSize(int bufferSize) {
		this.bufferSize = bufferSize;
	}

	@Override
	public void close() throws IOException {
		server.close();
	}

	static Integer count = 0;

	static void incre() {
		synchronized (count) {
			count++;
		}
	}

	@Override
	public void setReceiveHandler(IEndpointHandler handler) {
		this.handler = handler;
	}

	@Override
	public SocketAddress getLocalAddress() throws IOException {
		return server.getLocalAddress();
	}

	@Override
	public void openSession(SocketAddress target, Object attachment, SendHandler handler) {
		handler.onSession(new UdpEndpointSession(self(), target), attachment);
	}

	public static void main(String[] args) throws IOException, Exception {
		UdpEndpoint s = new UdpEndpoint();
		IEndpointHandler h = new IEndpointHandler() {
			@Override
			public void onAccept(EndpointSession ep) {
				System.out.println("a");
			}

			@Override
			public void onReceive(EndpointSession ep, Packet packet) {
//				UdpPacket p = UdpPacket.allocate(16, true);
//				p.put("Hello".getBytes());
//				try {
//					ep.send(p.flip());
//				} catch (IOException e) {
//					e.printStackTrace();
//				}
				incre();
			}

			@Override
			public void onFail(EndpointSession ep, Throwable ext) {
			}
		};
		s.setReceiveHandler(h);
		s.setBufferSize(1024);
		s.start(new InetSocketAddress("localhost", 10606));

		for (int i = 0; i < 10; i++) {
			final int num = i;
			Thread t = new Thread() {
				@Override
				public void run() {
					try {
						runClient(num);
					} catch (Exception e) {
						e.printStackTrace();
					}
				};
			};
			Thread.yield();
			t.start();
		}
		Thread.sleep(4000);

		System.out.println(count);
	}

	static void runClient(int num) throws Exception {
		for (int i = 0; i < 2; i++) {
			Endpoint s = new UdpEndpoint();
			s.setReceiveHandler(new EndpointHandlerAdapter() {
				@Override
				public void onReceive(EndpointSession ep, Packet packet) {
					System.err.println("client recv: " + packet.getLength());
				}
			});
			s.bind(new InetSocketAddress("localhost", 10607 + num + i * 1000));
			final Packet p = s.allocatePacket(1020);
			for (int j = 0; j < 1000; j++)
				p.put((byte) 'h');
			p.flip();
			s.openSession(new InetSocketAddress("localhost", 10606), null, new SendHandler() {
				@Override
				public void onSession(EndpointSession session, Object attachment) {
					try {
						session.send(p, null, new CompletionHandler<Integer, Object>() {
							@Override
							public void completed(Integer arg0, Object arg1) {
								//ByteBuffer b = p.buffer();
								//System.out.println("send: " + arg0 + ", " + b.limit() + "," + b.position());
							}
							@Override
							public void failed(Throwable arg0, Object arg1) {
								arg0.printStackTrace();
							}
						});
					} catch (IOException e) {
						e.printStackTrace();
					}
				}

				@Override
				public void onFailed(Throwable exc, Object attachment) {
					exc.printStackTrace();
				}
			});
		}
	}
	
	private static String genEndpointId(SocketAddress address)
	{
		InetSocketAddress i = (InetSocketAddress) address;
		return i.getAddress().getHostAddress()+":"+i.getPort();
	}
	@Override
	public Packet allocatePacket(int capacity) {
		return UdpPacket.allocate(capacity);
	}
	@Override
	public Packet wrapPacket(ByteBuffer src) {
		return UdpPacket.wrap(src);
	}
}

class ActUnit {
	Packet pack;
	SocketAddress target;
	Object attachment;
	CompletionHandler<Integer, ? super Object> handler;
	int ackTime = 3;
	int retry = 3;
}

class TimeUnit {
	static int MaxTime = 30;
	long lastSendTime;
	int lastPackId = -1;
	int avgTime = 3;
}

class UdpEndpointSession extends EndpointSession {
	private UdpEndpoint localEP;
	private InetSocketAddress remoteAddress;
	private Object attachment;
	private Packet dataBuffer;

	public UdpEndpointSession(UdpEndpoint localEP, SocketAddress remoteAddress) {
		this.localEP = localEP;
		this.remoteAddress = (InetSocketAddress) remoteAddress;
	}

	@Override
	public void send(Packet src, Object attachment, final CompletionHandler<Integer, Object> handler) {
		if (!src.validate())
			throw new RuntimeException("package formar error.");
		localEP.sendTo(src, remoteAddress, attachment, handler);
	}

	@Override
	public void send(Packet pack) {
		localEP.sendTo(pack, remoteAddress);
	}

	/**
	 * @return the dataBuffer
	 */
	public Packet getDataBuffer() {
		return dataBuffer;
	}

	/**
	 * @param dataBuffer
	 *            the dataBuffer to set
	 */
	public void setDataBuffer(Packet dataBuffer) {
		this.dataBuffer = dataBuffer;
	}

	/**
	 * @return the remoteAddress
	 */
	@Override
	public SocketAddress getRemoteAddress() {
		return remoteAddress;
	}

	/**
	 * @return the localAddress
	 * @throws IOException
	 */
	@Override
	public SocketAddress getLocalAddress() throws IOException {
		return (InetSocketAddress) localEP.getLocalAddress();
	}

	/**
	 * @return the attachment
	 */
	public Object getAttachment() {
		return attachment;
	}

	/**
	 * @param attachment
	 *            the attachment to set
	 */
	public void setAttachment(Object attachment) {
		this.attachment = attachment;
	}

	@Override
	public void close() {
		localEP = null;
		remoteAddress = null;
		dataBuffer = null;
		attachment = null;
	}

	@Override
	public void send(ByteBuffer buf) throws IOException {
		localEP.server.write(buf);
	}
}